package com.desk.api.mqtt;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;

import com.desk.api.service.LearnService;
import com.desk.api.service.TempHumService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@ConditionalOnProperty(name = "smartDesk.enabled", havingValue = "true")
public class MqttApplication implements InitializingBean, DisposableBean {

  @Resource TempHumService tempHumService;
  @Resource LearnService learnService;

  @Value("${smartDesk.broker}")
  private String broker;

  // TLS/SSL
  // String broker = "ssl://broker.emqx.io:8883";
  @Value("${smartDesk.username}")
  private String username;

  @Value("${smartDesk.password}")
  private String password;

  String clientid = "smartDesk_client";

  @Value("${smartDesk.temp-hum}")
  private String tempHumTopic;

  @Value("${smartDesk.learn}")
  private String learnTopic;

  MqttClient client = null;
  private static final ExecutorService executorService =
      new ThreadPoolExecutor(
          Runtime.getRuntime().availableProcessors(),
          Runtime.getRuntime().availableProcessors() * 2,
          60,
          TimeUnit.SECONDS,
          new LinkedBlockingQueue<>(50000));

  @Override
  public void destroy() throws Exception {
    try {
      if (client != null && client.isConnected()) {
        client.disconnect();
        client.close();
      }
      boolean awaitedTermination = executorService.awaitTermination(2, TimeUnit.SECONDS);
      if (!awaitedTermination) {
        executorService.shutdownNow();
      }
    } catch (Exception e) {
      log.error("mqtt close error", e);
    }
  }

  @Override
  public void afterPropertiesSet() throws Exception {
    client = new MqttClient(broker, clientid, new MemoryPersistence());
    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    int qos = 0;
    try {
      // 连接参数
      options.setConnectionTimeout(60);
      options.setKeepAliveInterval(60);
      // 设置回调
      client.connect(options);
      client.subscribe(
          tempHumTopic,
          qos,
          (topic, message) -> {
            String msg = new String(message.getPayload());
            log.info("topic: {},message: {}", topic, msg);
            executorService.execute(() -> tempHumService.reportData(msg));
          });
      client.subscribe(
          learnTopic,
          qos,
          (topic, message) -> {
            String msg = new String(message.getPayload());
            log.info("topic: {},message: {}", topic, msg);
            executorService.execute(() -> learnService.reportData(msg));
          });
    } catch (Exception e) {
      log.error("mqtt connect error", e);
    }
  }
}
